package cn.doitedu.rtmk.demo7;

import cn.doitedu.rtmk.beans.RuleMetaBean;
import cn.doitedu.rtmk.beans.UserEvent;
import cn.doitedu.rtmk.interfaces.IRuleCalculatorV3;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import groovy.lang.GroovyClassLoader;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.roaringbitmap.RoaringBitmap;

import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;

@Slf4j
public class RuleEngineCoreFunction7 extends KeyedBroadcastProcessFunction<Long, UserEvent, RuleMetaBean, JSONObject> {

    Map<Integer, IRuleCalculatorV3> ruleCalculatorMap = new HashMap<>();


    @Override
    public void open(Configuration parameters) throws Exception {
    }

    /**
     * 处理主流数据的方法
     * 静态圈选条件：月平均消费额>200的用户
     * 触发条件：当他发生A行为时（event_id='page_load', properties['url'] = '/page/001'），命中规则
     */
    @Override
    public void processElement(UserEvent userEvent, KeyedBroadcastProcessFunction<Long, UserEvent, RuleMetaBean, JSONObject>.ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {

        // 遍历运算机集合中的每一个规则运算机
        for (Map.Entry<Integer, IRuleCalculatorV3> entry : ruleCalculatorMap.entrySet()) {

            IRuleCalculatorV3 ruleCalculator = entry.getValue();
            // 传入当前的用户行为，交给规则运算机去处理
            ruleCalculator.calc(userEvent, out);

        }

    }

    /**
     * 处理广播流数据的方法
     */
    @Override
    public void processBroadcastElement(RuleMetaBean ruleMetaBean, KeyedBroadcastProcessFunction<Long, UserEvent, RuleMetaBean, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
        // 取出新收到的规则元信息中的： 规则id 和 预圈选人群bytes
        int rule_id = (int) ruleMetaBean.getRule_id();
        byte[] pre_select_users = ruleMetaBean.getPre_select_users();

        int rule_model_id = ruleMetaBean.getRule_model_id();
        String rule_param_json = ruleMetaBean.getRule_param_json();

        String rule_model_calc_groovy_code = ruleMetaBean.getRule_model_calc_groovy_code();


        // 反序列化出预圈选人群的bitmap对象
        RoaringBitmap preSelectUsersBitmap = RoaringBitmap.bitmapOf();
        ByteBuffer byteBuffer = ByteBuffer.wrap(pre_select_users);
        preSelectUsersBitmap.deserialize(byteBuffer);

        log.warn("收到一条新规则的元信息, 规则id:{},所属模型id:{},规则预圈选人群人数:{},规则参数:{}", rule_id, rule_model_id, preSelectUsersBitmap.getCardinality(), rule_param_json);


        /**
         * 利用groovy编译器，编译收到的规则所对应的规则运算机的代码
         * 并反射出实例对象
         * 然后传入预圈选人群、规则实例参数等信息来对规则运算机实例对象初始化
         */
        GroovyClassLoader groovyClassLoader = new GroovyClassLoader();
        Class aClass = groovyClassLoader.parseClass(rule_model_calc_groovy_code);
        IRuleCalculatorV3 calculator = (IRuleCalculatorV3) aClass.newInstance();
        calculator.open(preSelectUsersBitmap,JSON.parseObject(rule_param_json),getRuntimeContext());

        // 将规则运算机放入运算机集合
        ruleCalculatorMap.put(rule_id,calculator);

    }

}
